package com.raylu.realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.raylu.realtime.app.func.ConvertTimeStampFunction;
import com.raylu.realtime.app.func.ConvertTimeStrFunction;
import com.raylu.realtime.app.template.DimAsyncFunction;
import com.raylu.realtime.bean.OrderInfo;
import com.raylu.realtime.bean.PageBean;
import com.raylu.realtime.bean.ProvinceStats;
import com.raylu.realtime.utils.ClickHouseUtil;
import com.raylu.realtime.utils.KafkaSourceUtil;
import com.raylu.realtime.utils.PropertiesUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * Description:
 * <p>
 * Create by lucienoz on 2022/1/9.
 * Copyright © 2022 lucienoz. All rights reserved.
 */
public class ProvinceStatsApp2 {
    public static void main(String[] args) throws Exception {
        Properties load = PropertiesUtil.load("config.properties");
        //TODO 1. 准备运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2. 配置检查点
//        env.enableCheckpointing(5000);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        env.setStateBackend(new FsStateBackend(load.getProperty("province.stats.app.fsstatebackend.url")));
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, org.apache.flink.api.common.time.Time.seconds(10L), org.apache.flink.api.common.time.Time.hours(10L)));
//
        //TODO 3. 准备Flink表运行环境
        SingleOutputStreamOperator<PageBean> pageBeanDS = env.addSource(KafkaSourceUtil
                        .getKafkaSource(load.getProperty("province.stats.app.kafka.source-topic1"), "test"))
                .map(r -> JSONObject.parseObject(r, PageBean.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<PageBean>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<PageBean>() {
                            @Override
                            public long extractTimestamp(PageBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }));
        SingleOutputStreamOperator<PageBean> uvLogDS = env.addSource(KafkaSourceUtil
                        .getKafkaSource(load.getProperty("province.stats.app.kafka.source-topic2"), "test"))
                .map(r -> JSONObject.parseObject(r, PageBean.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<PageBean>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<PageBean>() {
                            @Override
                            public long extractTimestamp(PageBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }));
        SingleOutputStreamOperator<OrderInfo> orderInfoDS = env.addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("province.stats.app.kafka.source-topic3"), "test")).map(r -> JSONObject.parseObject(r, OrderInfo.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderInfo>() {
                            @Override
                            public long extractTimestamp(OrderInfo element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }));

        SingleOutputStreamOperator<PageBean> orderInfo = orderInfoDS.map(r -> {
            PageBean pageBean = new PageBean();
            pageBean.setAr(r.getId());
            pageBean.setDuring_time(r.getCreate_time());
            return pageBean;
        }).returns(PageBean.class);//Types.POJO(PageBean.class)


        DataStream<PageBean> union = pageBeanDS.union(uvLogDS, orderInfo);

        pageBeanDS.print("pageBeanDS");
        uvLogDS.print("uvLogDS");
        orderInfoDS.print("orderInfoDS");
        union.print("union");

        SingleOutputStreamOperator<PageBean> process = union.keyBy(r -> r.getAr())
                .window(TumblingEventTimeWindows.of(Time.seconds(3L)))
                .process(new ProcessWindowFunction<PageBean, PageBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<PageBean, PageBean, String, TimeWindow>.Context context, Iterable<PageBean> elements, Collector<PageBean> out) throws Exception {
                        elements.forEach(out::collect);
                    }
                });

        process.print("process");

        env.execute();

    }
}
